package cn.tedu.knows.kafka.demo;

import cn.tedu.knows.kafka.model.Message;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class Consumer {

    //用于将json格式字符串转换为java对象
    private Gson gson=new Gson();

    //Kafka消息的接收方接收信息的原理是靠Kafka依赖提供的"监听器"
    //监听器会一直关注指定的话题名称,如果该话题名称中出现消息,则开始接收

    @KafkaListener(topics = "myTopic")
    //下面的方法就是当指定话题接收到消息时要运行的代码
    public void receive(
            ConsumerRecord<String,String> record){
        //ConsumerRecord是kafka依赖提供的类型
        //表示从kafka获得的消息
        //<String,String>也是<[话题名称],[消息内容]>
        //获得消息内容
        String json=record.value();
        //转换为java对象
        Message message=gson.fromJson(json,Message.class);
        log.debug("接收方接收到消息:{}",message);
    }
}
